package day4

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.Row.empty.schema
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.udf

import java.text.SimpleDateFormat
import java.util.TimeZone

object time_consume {
  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir", "D:\\devtools\\hadoop")
    Logger.getLogger("org").setLevel(Level.OFF)
    val session = SparkSession.builder()
      .appName("SparkSQL-time_consume")
      .master("local")
      .getOrCreate()

    //生成一个表
    import session.implicits._
    session.sparkContext.textFile("D:\\data\\files\\time_consume.log")
      .map( i =>{
        val list = i.split("\t")
        (list(1),list(2).toLong,list(3))
      })
      .toDF("name","time","amount")
      .createOrReplaceTempView("time_consume")

    //基础解法
    //    session.sql("select name , " +
    //      "concat_ws(',', collect_list(CAST(time AS STRING))) AS times ," +
    //      "round(sum(amount),2) as total " +
    //      "from time_consume " +
    //      "group by name").show()

    //UDF
    // 创建一个UDF
    //    val timeChange = udf((time: Long) =>{
    //      val nyr = new SimpleDateFormat("yyyy-MM-dd")
    //      nyr.setTimeZone(TimeZone.getTimeZone("CST"))
    //      nyr.format(time * 1000)
    //    })
    session.udf.register("timestamp_to_string",(i:Long)=> {
      val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
      sdf.setTimeZone(TimeZone.getTimeZone("CST"))
      sdf.format(i*1000)
    })
    //在SparkSQL当中注册这个方法
    //    session.udf.register("timeChange",timeChange)

    //创建一个SQL并使用新建的UDF方法
    val sql = "select name," +
      "concat_ws(',',collect_list(timestamp_to_string(time))) AS times," +
      "round(sum(amount),2) as total " +
      "from time_consume group by name"
    session.sql(sql).show(30)

    //直接使用DF进行数据捕获
    //format:text,csv,json,jdbc
    //    session.read.format("text")
    //      .option("tab","\t")
    //      .schema(schema)
    //      .csv("D:\\data\\files\\time_consume.log")
    //      .createOrReplaceTempView("time_consume")
    //    session.sql("select * from time_consume").show()

  }
}